package com.doit.doit47;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.sql.*;
import java.time.Duration;
import java.util.Arrays;
import java.util.Timer;
import java.util.TimerTask;

public class _08_练习之消费者之Mysql {

    public static void main(String[] args) throws SQLException {
        KafkaConsumer<String, String> consumer = MyKafkaUtils.getConsumer("aaa", false, "earliest");

        consumer.subscribe(Arrays.asList("event-log"));
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test01", "root", "123456");
        PreparedStatement pps = conn.prepareStatement("insert into guid values(?)");
        PreparedStatement get_uv = conn.prepareStatement("select count(distinct guid) as res from guid");
        new Thread(new MyThread(consumer,pps)).start();

        Timer timer = new Timer();
        timer.schedule(new MyTask(get_uv),10000,5000);
    }
}

class MyThread implements Runnable{
    private KafkaConsumer<String, String> consumer;
    private PreparedStatement pps;
    public MyThread(KafkaConsumer<String, String> consumer,PreparedStatement pps){
        this.consumer = consumer;
        this.pps = pps;
    }
    @Override
    public void run() {
        while (true){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
            for (ConsumerRecord<String, String> record : poll) {
                String json = record.value();
                EventLog eventLog = JSON.parseObject(json, EventLog.class);
                int guid = eventLog.getGuid();
                try {
                    pps.setInt(1,guid);
                    pps.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

class MyTask extends TimerTask{
    private PreparedStatement get_uv;
    public MyTask(PreparedStatement get_uv){
        this.get_uv = get_uv;
    }
    @Override
    public void run() {
        ResultSet resultSet = null;
        try {
            resultSet = get_uv.executeQuery();
            while (resultSet.next()){
                long uv = resultSet.getLong(1);
                System.out.println("当前的uv是："+uv);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
